/**
 * Copyright 2007-2008. Chongqing First Information & Network Co., Ltd. All
 * rights reserved. <a>http://www.cqfirst.com.cn</a>
 */
package com.cqcis.uip.base.framework.utils;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 同步队列, 不限制容量
 * 
 * @author huangmy
 * @date 2007-12-13
 */
public class ConcurrentTaskQueue<T> implements Serializable {
	private static final long serialVersionUID = 1L;
	private Queue<T> taskQueue = new ConcurrentLinkedQueue<T>();
	private boolean notifyEmpty;// 空
	private boolean waitStated; // 等待状态
	private List<EmptyBlockListener> listeners = new ArrayList<EmptyBlockListener>();//方法阻塞通知
    private List<PolledListener> polledListeners = new ArrayList<PolledListener>();// polled
    private List<OfferedListener> offeredListeners = new ArrayList<OfferedListener>();// offered
    private List<OfferingListener> offeringListeners = new ArrayList<OfferingListener>();// offering
	
	/**
	 * 唤醒在queue上阻塞等待的线程
	 * 
	 */
	public void notifyQueue() {
		synchronized (taskQueue) {
			if (waitStated) {
				notifyEmpty = true;
			}
			
		    taskQueue.notify();
		}
	}
	
	/**
	 * 添加任务ID， 同时唤醒一个等待该任务ID的线程
	 * @param taskId
	 * @return
	 */
	public boolean offer(T task) {
		boolean offered = false;
		fireOfferingListeners(task);
		synchronized (taskQueue) {
			fireOfferedListeners(task);
			offered = taskQueue.offer(task);
			taskQueue.notify();
		}
		
		return offered;
	}
	
	/**
	 * 增加所有
	 * 
	 * @param tasks
	 * @return
	 */
	public boolean addAll(Collection<T> tasks) {
		boolean s = false;
		
		for (Iterator<T> it = tasks.iterator(); it.hasNext();) {
			T task = it.next();
			
			s = offer(task);			
			if (!s) {
				return false;
			}
		}
		
		return true;
	}
	
	/**
	 * 在队首添加任务
	 * 
	 * @param task
	 * @return
	 */
	public boolean addFirst(T task) {
		boolean offered = false;
		fireOfferingListeners(task);
		synchronized (taskQueue) {
			Queue<T> taskQueueTemp = new ConcurrentLinkedQueue<T>();
			taskQueueTemp.addAll(taskQueue);
			
			taskQueue.clear();
			
			offered = taskQueue.offer(task);
			
			taskQueue.addAll(taskQueueTemp);
			
			taskQueue.notify();
		}
		
		if (task != null && offered) {
			fireOfferedListeners(task);
		}

		return offered;
	}
    
	/**
	 * 移除一个任务ID, 如果没有可用的任务ID, 调用该方法的线程将阻塞
	 * @return
	 */
	public T poll() {		
		return poll(true, 0);
	}
	
	/**
	 * 带有超时的poll, 如果在指定的时间内没有获取任务, 该方法返回 null
	 * @param timeout  - 毫秒; 0 不超时
	 * @return
	 */
	public T poll(long timeout) {
		return poll(true, timeout);
	}
	
	/**
	 * 移除一个任务ID
	 * @param block - 如果没有可用的任务ID, 调用该方法的线程是否阻塞
	 * @return
	 */
	public T poll(boolean block, long timeoutIfBlock) {		
		T task = null;

		while (true) {
			synchronized (taskQueue) {
				if (((task = taskQueue.poll()) == null) && block) {
					try {
						fireBlockingListeners();// will blocking event
						
						waitStated = true;
						if (timeoutIfBlock == 0) {
							taskQueue.wait();
						} else {
						    taskQueue.wait(timeoutIfBlock);	
						    block = false;
						}
						
						if (notifyEmpty) {
							notifyEmpty = false;
							break;
						}
					} catch (InterruptedException e) {
						break;
					} finally {
						waitStated = false;
					}
				} else {
					break;
				}
			}
		}
		
		if (task != null) {
			firePolledListeners(task);
		}

		return task;
	}
	
	/**
	 * 发布事件
	 * 
	 */
	private void fireBlockingListeners() {
		for (EmptyBlockListener l : this.listeners) {
			try {
			    l.blocking();
			} catch (Throwable t) {				
			}
		}
	}
	
	/**
	 * 发布节点被移除事件
	 * 
	 * @param t
	 */
	private void firePolledListeners(T t) {
		for (PolledListener l : this.polledListeners) {
			try {
				l.polled(t);
			} catch (Throwable e) {
			}
		}
	}
	
	/**
	 * 发布节点被植入事件
	 * 
	 * @param t
	 */
	private void fireOfferedListeners(T t) {
		for (OfferedListener l : this.offeredListeners) {
			try {
				l.offered(t);
			} catch (Throwable e) {
			}
		}
	}
	
	/**
	 * 发布节点将植入事件
	 * 
	 * @param t
	 */
	private void fireOfferingListeners(T t) {
		for (OfferingListener l : this.offeringListeners) {
			try {
				l.offering(t);
			} catch (Throwable e) {
			}
		}
	}
	
	/**
	 * 添加队列为空监听器
	 * 
	 * @param l
	 */
	public void addEmptyBlockListener(EmptyBlockListener l) {
		this.listeners.add(l);
	}
	
	/**
	 * 添加节点被移除监听器
	 * 
	 * @param l
	 */
	public void addPolledListener(PolledListener l) {
		this.polledListeners.add(l);
	}
	
	/**
	 * 添加节点被植入监听器
	 * 
	 * @param l
	 */
	public void addOfferedListener(OfferedListener l) {
		this.offeredListeners.add(l);
	}
	
	/**
	 * 添加节点将植入监听器
	 * 
	 * @param l
	 */
	public void addOfferingListener(OfferingListener l) {
		this.offeringListeners.add(l);
	}
	
	/**
	 * 获取一个任务ID, 但不移除, 如果没有可用的任务ID, 该方法返回null
	 * @return
	 */
	public T peek() {		
		return taskQueue.peek();
	}
	
	/**
	 * 移除一个任务ID, 如果没有可用的任务ID, 该方法返回null
	 * @return
	 */
	public T remove() {
		return poll(false, 0);
	}
	
	/**
	 * 删除指定
	 * 
	 * @param obj
	 * @return
	 */
	public boolean remove(T obj) {
		return taskQueue.remove(obj);
	}
	
	/**
	 * 包含
	 * 
	 * @param obj
	 * @return
	 */
	public boolean contains(Object obj) {
		return taskQueue.contains(obj);
	}
	
	/**
	 * 当前队列中的任务数量
	 * @return
	 */
	public int size() {
		return taskQueue.size();
	}
	
	/**
	 * 监听队列为空，且取数据方法阻塞
	 * 
	 * @author huangmy
	 * @date   2008-9-10
	 */
	public static interface EmptyBlockListener extends Serializable {
		/**
		 * 将阻塞
		 * 
		 */
		public void blocking();
	}
	
	/**
	 * 节点元素被取出
	 * 
	 * @author huangmy
	 * @date   2008-9-12
	 */
	public static interface PolledListener {
		/**
		 * polled
		 * 
		 * @param t
		 */
		public void polled(Object t);
	}
	
	/**
	 * 节点元素被植入
	 * 
	 * @author huangmy
	 * @date   2008-9-12
	 */
	public static interface OfferedListener {
		/**
		 * offered
		 * 
		 * @param t
		 */
		public void offered(Object t);
	}
	
	/**
	 * 节点元素将植入
	 * 
	 * @author huangmy
	 * @date   2008-9-12
	 */
	public static interface OfferingListener {
		/**
		 * offered
		 * 
		 * @param t
		 */
		public void offering(Object t);
	}
}
